package com.ysw.java_api;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.io.IOException;
import java.time.Duration;
import java.util.*;

/**
 * @ClassName MyConsumer
 * @Description TODO
 * @Author ysw
 * @Date 2021/12/12 21:15
 */
public class MyConsumer {

    private static final String TOPIC = "first";

    public static void main(String[] args) throws IOException {
        Properties properties = new Properties();
        // 服务器信息
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.37.11:9092");
        // 是否自动提交
        /*
         * 自动提交有可能会丢数据，自动提交时机难以把握，并不知道是否已经处理完毕，如果自动提交，然而程序出问题了，数据没有操作成功，这批数据处理就丢弃了
         * 关闭自动提交之后，有程序员自主手动提交，手动提交有俩种方式。手动提交会导致数据重复
         * 1. 同步提交
         * 2. 异步提交
         * 手动提交如果出现异常怎么办？怎么更好的去处理offset
         */
        //properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自动提交延时
        //properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        /*
         *  重置消费者的offest,默认是latest
         *  有俩个可选值  earliest latest，触发条件，数据删除，换了组
         *  latest: 每次从最大的offset开始消费，以前的消费数据无法消费到
         *  earliest：每次从最早的offset开始消费
         */
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 定义消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test_group");

        // consumer 给broker 发送心跳间隔时间
        properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);

        // kafka 如果超过10s没有收到消费者的心跳，进行rebalance,把分区分配给其他消费者
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

        // 一次poll 最大拉取的消息条数，可以根据消费速度的快慢来进行设置
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);

        // 如果俩次poll的时间超过30s 的时间间隔，kafka会认为消费者消费能力弱，将其踢出消费者组。将分区分配给其他消费者
        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        // 消费方式
        /*
           1. 直接进行消费     consumer.subscribe(Arrays.asList(TOPIC));
           2. 指定分区进行消费
         */
        consumer.subscribe(Arrays.asList(TOPIC));
        //1.指定分区进行消费
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC,0)));
        // 2.消息回溯消费,从开始进行消费
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC,0)));
        //consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC,0)));
        // 3.指定offset进行消费
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC,0)));
        //consumer.seek(new TopicPartition(TOPIC,0),10);
        // 4.从指定时间进行消费
        /*List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC);
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        HashMap<TopicPartition, Long> map = new HashMap<>();
        partitionInfos.forEach(e-> map.put(new TopicPartition(TOPIC,e.partition()),fetchDataTime));
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        parMap.entrySet().forEach(entry->{
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if(null == key || null ==value){
                return;
            }
            if(value != null){
                long offset = value.offset();
                consumer.assign(Arrays.asList(key));
                consumer.seek(key,offset);
            }
        });*/
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            records.forEach(e -> System.out.println(e.value() + "--" + e.offset()));
            if (records.count() > 0) {
                consumer.commitSync(); // - 阻塞-提交成功
              /*  // 异步手动提交
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        if(null != e){
                            // 产生异常，处理自己的业务逻辑
                        }
                    }
                });*/
            }
            // 指定分区消费
            // 消息回溯消费

        }
    }
}